package cn.cdqf.coupon.dispatch.kafka;

import cn.cdqf.coupon.common.CommonConstant;
import cn.cdqf.coupon.dispatch.dao.CouponCodeUserMapper;
import cn.cdqf.coupon.dispatch.service.CouponCodeUserService;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Maps;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Component
public class DispatchKafkaReceiver {

    @Autowired
    private CouponCodeUserService couponCodeUserService;
    @Autowired
    private ObjectMapper objectMapper;
    @KafkaListener(topics = {CommonConstant.KafkaKey.COUPONID_MERCHANTID_TO_CODE},groupId = "test-consumer-group")
    public void listen(ConsumerRecord<?, String> record) throws IOException {
        String message = record.value();
        HashMap<String,String> hashMap = objectMapper.readValue(message, HashMap.class);
        //商家id
        Set<String> set = hashMap.keySet();
        for (String s : set) {
            HashMap<String,String> cidForCode = objectMapper.readValue(hashMap.get(s), HashMap.class);
            for(Map.Entry<String,String> entry:cidForCode.entrySet()){
                List<String> list = objectMapper.readValue(entry.getValue(), List.class);
                couponCodeUserService.batchInsert(list,s,entry.getKey());
            }
        }
        System.out.println("收到kafka生产者发送的消息"+record.value());
    }
}
